package com.iflyteck.client;

import com.iflyteck.coder.RpcEncoder;
import com.iflyteck.handler.ServerInfo;
import com.iflyteck.handler.UserClientHandler;
import com.iflyteck.json.JSONSerializer;
import com.iflyteck.pojo.RpcRequest;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.io.InputStream;
import java.util.*;
import java.util.stream.Collectors;

public class ClientUtils {

    public static Map<String, UserClientHandler> userClientHandlerMap = new HashMap<>();

    public static void loadZkProvider() throws Exception {
        // 加载配置文件
        InputStream resource = ClientUtils.class.getClassLoader().getResourceAsStream("application.properties");
        Properties properties = new Properties();
        properties.load(resource);
        String zkServerConnectUrl = properties.getProperty("zdy.zkServerConnectUrl");
        String serverBaseNode = properties.getProperty("zdy.serverBaseNode");
        // 连接zookeeper服务
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory
                .builder().connectString(zkServerConnectUrl)
                .sessionTimeoutMs(1000)
                .retryPolicy(retryPolicy)
                .build();
        client.start();
        // 获取server节点
        List<String> serverNodeList = client.getChildren().forPath(serverBaseNode);
        for (String serverNodePath : serverNodeList) {
            System.out.println("获取数据的路径：" + serverBaseNode + "/" + serverNodePath);
            byte[] bytes = client.getData().forPath(serverBaseNode + "/" + serverNodePath);
            String[] nodeInfo = new String(bytes).split(":");
            // 初始化连接器
            buildClientHandler(nodeInfo[0],nodeInfo[1]);
        }
        // 添加根节点的变更监听器，监听server服务器的加入和下线
        PathChildrenCache cache = new PathChildrenCache(client, serverBaseNode, true);
        cache.start();
        PathChildrenCacheListener listener = (server,event) -> {
            if (PathChildrenCacheEvent.Type.CHILD_ADDED.equals(event.getType())) {
                // 检测到新节点加入，把新的server加入到处理器集合中
                String[] serverNodeInfo = new String(event.getData().getData()).split(":");
                System.out.println("有新的server节点加入：" + serverNodeInfo[0] + ":" + serverNodeInfo[1]);
                buildClientHandler(serverNodeInfo[0],serverNodeInfo[1]);
            } else if (PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(event.getType())) {
                // 检测到节点删除，把server从处理器集合中移除
                String[] serverNodeInfo = new String(event.getData().getData()).split(":");
                System.out.println("server节点下线:" + serverNodeInfo[0] + ":" + serverNodeInfo[1]);
                userClientHandlerMap.remove(serverNodeInfo[0] + serverNodeInfo[1]);
            }
        };
        cache.getListenable().addListener(listener);

    }

    private static void buildClientHandler(String ip, String port) throws InterruptedException {
        ServerInfo serverInfo = new ServerInfo(ip, Integer.parseInt(port));
        UserClientHandler userClientHandler = initClient(serverInfo);
        userClientHandlerMap.put(ip + port, userClientHandler);
    }


    // 3、初始化客户端（创建连接池bootstrap  设置bootstrap  连接服务器）
    public static UserClientHandler initClient(ServerInfo serverInfo) throws InterruptedException {
        // 1)初始化UserClientHandler
        final  UserClientHandler userClientHandler = new UserClientHandler();
        userClientHandler.setServerInfo(serverInfo);
        // 2)创建连接池对象
        NioEventLoopGroup group = new NioEventLoopGroup();
        // 3)创建客户端的引导对象
        Bootstrap bootstrap = new Bootstrap();
        // 4)配置引导类对象
        bootstrap.group(group)
                //设置通道为Nio
                .channel(NioSocketChannel.class)
                // 设置请求协议为TCP
                .option(ChannelOption.TCP_NODELAY,true)
                // 监听channel并初始化
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //1、获取ChannelPipeline
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        //2、设置编码 解码
                        pipeline.addLast(new RpcEncoder(RpcRequest.class,new JSONSerializer()));
                        pipeline.addLast(new StringDecoder());
                        //3、添加自定义时间处理器
                        pipeline.addLast(userClientHandler);
                    }
                });

        // 5、连接服务器
        bootstrap.connect(serverInfo.getIp(),serverInfo.getPort()).sync();
        return userClientHandler;
    }

    public static UserClientHandler getAvailableUserClientHandler() {
        if (userClientHandlerMap.isEmpty()) {
            return null;
        }
        List<ServerInfo> servers = new ArrayList<>();
        for (String ipPort : userClientHandlerMap.keySet()) {
            ServerInfo serverInfo = userClientHandlerMap.get(ipPort).getServerInfo();
            // 当前时间
            long currentTime = System.currentTimeMillis();
            // 判断最后一次响应时间是否超过5秒，超过5秒则清零
            if (currentTime - serverInfo.getLastCallTime() >= 5000) {
                serverInfo.setSpendTime(0);
            }
            servers.add(serverInfo);
        }
        // 按照响应时间排序，选取时间最短的server
        List<ServerInfo> sortServerList = servers.stream().
                sorted(Comparator.comparing(ServerInfo::getSpendTime))
                .collect(Collectors.toList());

        ServerInfo serverInfo = sortServerList.get(0);
        UserClientHandler handler = userClientHandlerMap.get(serverInfo.getIp() + serverInfo.getPort());
        return handler;

    }
}
